/*
 * Copyright 2006 by RidgeSoft, LLC., PO Box 482, Pleasanton, CA  94566, U.S.A.
 * www.ridgesoft.com
 *
 * RidgeSoft grants you the right to use, modify, make derivative works and
 * redistribute this source file provided you do not remove this copyright notice.
 */

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
//import java.io.PrintStream;

/**
 * This class implements a reliable means for passing messages over an unrelaible
 * serial data stream, such as a serial port. The protocol supports messages as
 * short arrays of binary data.  It does not restrict the data which messages may
 * contain, although the messages must be shorter than the specified receive length.
 * The protocol ensures the messages are delivered to the receiver in the order
 * they are transmitted.  The protocol also implements flow control, ensuring the
 * transmitter cannot send messages faster than the output stream can deliver
 * them and the receiver can process them.
 * <p>
 * The protocol is symmetric, allowing both ends of the stream to use this class
 * to pass data.
 */
public class Mailbox {
    private static final int STATE_CLOSED           = 0;
    private static final int STATE_OPEN_SENT        = 1;
    private static final int STATE_OPEN_RECEIVED    = 2;
    private static final int STATE_OPENED           = 3;
    private static final int STATE_CLOSING          = 4;

    // framing and escape characters
    // special characters < 0x80 to avoid sign extension issues
    private static final byte FRAME         = 0x7c;
    private static final byte ESC           = 0x7d;
    private static final byte ESC_FRAME     = 0x7e;
    private static final byte ESC_ESC       = 0x7f;

    // control byte definitions
    // Note: FRAME and ESC cannot be used as control bytes
    private static final byte OPEN      = 1;
    private static final byte OPEN_ACK  = 2;
    private static final byte CLOSED    = 3;
    // Control bytes with the ESTABLISHED_BIT most significant
    // bit set are ACKs with or without data.
    //
    // ACK - 0b10wwaaii
    //          ww - window close id
    //          aa - acknowledge id
    //          ii - data message id, ignore if no data
    private static final int ESTABLISHED_BIT = 0x80;

    private static final int BUFFER_SIZE = 60;

    private static byte[] EMPTY_MESSAGE = new byte[0];

//    private PrintStream mDebugStream = System.out;

    private InputStream mInputStream;
    private OutputStream mOutputStream;
    private int mRetransmitTime = 1000;         // minimum delay before retransmit
    private int mRxSleepTime = 4;               // 4 milliseconds, less than 64 bytes @ 115.2K
    private int mReceiveErrors = 0;
    private int mRetransmitCount = 0;

    private int mBasePriority;

    private byte[][] mTxQueue = new byte[4][];
    private boolean[] mRetransmit = new boolean[4];
    private byte mLastTxControlValue;
    private int mNextTx;
    private int mNextPost;
    private int mTxWindowClosedId;
    private int mAckId;
    private int mLastTxAckd;

    private int mMaxRxLength;
    private byte[][] mRxQueue = new byte[4][];
    private int mNextRx;
    private int mNextGet;
    private int mRxWindowClosedId;

    private byte[] mTxBuffer = new byte[BUFFER_SIZE+4];     // extra space to elimante special cases
    private byte[] mRxBuffer = new byte[BUFFER_SIZE];
    private int mRxBufferCount;
    private int mRxBufferIndex;

    private int mState = STATE_CLOSED;

    private ReceiveThread mReceiveThread;
    private SendThread mSendThread;
    private RetransmitThread mRetransmitThread;

    /**
     * Constructs a Mailbox object.
     * <p>
     * @param inputStream
     *              an InputStream object
     * @param outputStream
     *              an OutputStream object
     * @param maximumReceiveLength
     *              a value as large or larger than the maximum length of any receive message.
     *              Larger messages will be discarded.
     * @param basePriority
     *              the base priority for the mailbox's threads.  The mailbox will create threads
     *              which run at this priority and the next higher priority when it is openned.
     */
    public Mailbox(InputStream inputStream, OutputStream outputStream, int maximumReceiveLength, int basePriority) {
        mInputStream = inputStream;
        mOutputStream = outputStream;
        mMaxRxLength = maximumReceiveLength;
        mBasePriority = basePriority;
    }

    /**
     * Open the mailbox, waiting as long as necessary to connect with the peer at the other
     * end of the streams.
     *
     * @throws IOException
     */
    public void open() throws IOException {
        open(0);
    }

    /**
     * Open the mailbox, waiting no longer than the timeout value to connect with the peer
     * at the other end of the streams.
     *
     * @param timeout
     *              timeout in milliseconds
     * @throws IOException
     */
    public void open(int timeout) throws IOException {
        if (mState != STATE_CLOSED)
            throw new MailboxStateException("Not closed");

        resetQueues();

        mState = STATE_OPEN_SENT;

        mAckId = 0;
        mRxWindowClosedId = 0;
        mLastTxAckd = 0;

        mRxBufferCount = 0;
        mRxBufferIndex = 0;

        mReceiveThread = new ReceiveThread();
        mReceiveThread.setPriority(mBasePriority+1);
        mReceiveThread.setDaemon(true);
        mReceiveThread.start();

        mSendThread = new SendThread();
        mSendThread.setPriority(mBasePriority);
        mSendThread.setDaemon(true);
        mSendThread.start();

        mRetransmitThread = new RetransmitThread();
        mRetransmitThread.setPriority(mBasePriority);
        mRetransmitThread.setDaemon(true);
        mRetransmitThread.start();

        synchronized (this) {
            try {
                if (mState != STATE_OPENED)
                    wait(timeout);
            }
            catch (InterruptedException e) {}
        }
        if (mState != STATE_OPENED) {
            close();
            throw new MailboxStateException("Open failed");
        }
    }

    private void updateState(int newState) {
        synchronized (this) {
            mState = newState;
            if ((newState == STATE_OPENED)  || (newState == STATE_CLOSED))
                notifyAll();
        }
        synchronized (mTxQueue) {
            mLastTxControlValue = 0;
            mTxQueue.notifyAll();
        }
    }

    private void ack() {
        synchronized (mTxQueue) {
            mLastTxControlValue = 0;
            mTxQueue.notifyAll();
        }
    }

    /**
     * Close the mailbox.
     */
    public void close() {
        if (mState == STATE_CLOSED || mState == STATE_CLOSING)
            return;

        updateState(STATE_CLOSING);

        synchronized (mTxQueue) {
            mTxQueue.notifyAll();
        }
        synchronized (mRxQueue) {
            mRxQueue.notifyAll();
        }

        synchronized (this) {
           while (mState != STATE_CLOSED) {
                try {
                    wait();
                }
                catch (InterruptedException e) {}
            }
        }

        if (mRetransmitThread != null) {
            mRetransmitThread.interrupt();
            try {
                mRetransmitThread.join();
            }
            catch (InterruptedException e) {}
            mRetransmitThread = null;
        }

        if (mSendThread != null) {
            try {
                mSendThread.join();
            }
            catch (InterruptedException e) {}
            mSendThread = null;
        }

        if (mReceiveThread != null) {
            try {
                mReceiveThread.join();
            }
            catch (InterruptedException e) {}
            mReceiveThread = null;
        }

        resetQueues();
    }

    private void resetQueues() {
        // reset the transmitter
        synchronized (mTxQueue) {
            for (int i =0; i < mTxQueue.length; ++i) {
                mTxQueue[i] = null;
                mRetransmit[i] = false;
            }
            mNextTx = 1;
            mNextPost = 1;
            mTxQueue.notifyAll();
        }

        // reset the receiver
        synchronized (mRxQueue) {
            for (int i =0; i < mRxQueue.length; ++i)
                mRxQueue[i] = null;
            mNextRx = 1;
            mNextGet = 1;
            mRxQueue.notifyAll();
        }
    }

    /**
     * Sets the time the receiver thread sleeps between polls
     * of the input stream when there is no data.  This time must be
     * small enough that the input stream's buffer does not get
     * overrun.
     *
     * @param rxSleepTime
     *              time in milliseconds to wait between receive polls.
     *              The default is 4 milliseconds.
     */
    public void setRxSleepTime(int rxSleepTime) {
        mRxSleepTime = rxSleepTime;
    }

    /**
     * Sets the amount of time to wait before retransmitting an
     * unacknowledged message.
     *
     * @param rxTime
     *              retransmit timeout in milliseconds.
     *              The default is 1000 milliseconds.
     */
    public void setRetransmitTime(int rxTime) {
        mRetransmitTime = rxTime;
    }

    /**
     * Gets the count of recieve message errors.
     *
     * @return
     *              receive error count
     */
    public int getReceiveErrors() {
        return mReceiveErrors;
    }

    /**
     * Gets the count of retransmissions.
     *
     * @return
     *              count of retransmits
     */
    public int getRetransmitCount() {
        return mRetransmitCount;
    }

    /**
     * Posts a message to be delivered to the peer.  This method does not wait for
     * the peer to acknowledge receipt of the message.  This method will block if the
     * transmit queue is full, which prevents the peer's receiver from being overrun.
     * This method does not timeout when it blocks.
     *
     * @param message
     *              a byte array containing the message to send
     * @throws IOException
     */
    public void postMessage(byte[] message) throws IOException {
        postMessage(message, 0);
    }

    /**
     * Posts a message to be delivered to the peer.  This method does not wait for
     * the peer to acknowledge receipt of the message.  This method will block if the
     * transmit queue is full, which prevents the peer's receiver from being overrun.
     * This method has a timeout to control how long it will wait when blcoked.
     *
     * @param message
     *              a byte array containing the message to send
     * @param maximumWaitTime
     *              the maximum time in milliseconds to wait.  The value 0 indicates
     *              do not timeout.
     * @throws IOException
     */
    public void postMessage(byte[] message, int maximumWaitTime) throws IOException {
        long waitUntil = 0;
        synchronized (mTxQueue) {
            while (mTxQueue[mNextPost] != null) {
                try {
                    if (maximumWaitTime == 0)
                        mTxQueue.wait();
                    else {
                        // the thread wakes up on any transmit queue event, so timing
                        // is a bit more complicated
                        if (waitUntil == 0)
                            waitUntil = System.currentTimeMillis() + maximumWaitTime;

                        long thisWait = waitUntil - System.currentTimeMillis();
                        if (thisWait > 0)
                            mTxQueue.wait(thisWait);
                        else
                            throw new IOException("Timeout");
                    }
                    if (mState != STATE_OPENED)
                        throw new MailboxStateException();
                }
                catch (InterruptedException e) {}
            }
            if (mState != STATE_OPENED)
                throw new MailboxStateException();
            mTxQueue[mNextPost] = message;
            mNextPost++;
            mNextPost &= 0x3;         // limit queue to 4 entries
            mTxQueue.notifyAll();
        }
    }

    private void sendMessage(byte[] message, byte control) throws IOException {
/*
            mDebugStream.print("TX: " + Integer.toHexString(control & 0xff));
            if ((control & ESTABLISHED_BIT) != 0) {
                System.out.print(
                    ", w: " + ((control >> 4) & 0x03) +
                    ", a: " + ((control >> 2) & 0x03) +
                    ", i: " + (control & 0x03));
            }

            if (message.length > 0) {
                mDebugStream.print(", m: ");
                for (int i = 0; i < message.length; ++i)
                    mDebugStream.print(Integer.toHexString(message[i] & 0xff) + ' ');
            }
            mDebugStream.println();
        }
*/
        mLastTxControlValue = control;
        mTxBuffer[0] = FRAME;
        mTxBuffer[1] = control;
        int offset = 2;
        int checksum = mTxBuffer[1];
        for (int i = 0; i < message.length; ++i) {
            int data = message[i];
            checksum += data;
            switch (data) {
            case FRAME:
                mTxBuffer[offset++] = ESC;
                mTxBuffer[offset++] = ESC_FRAME;
                break;
            case ESC:
                mTxBuffer[offset++] = ESC;
                mTxBuffer[offset++] = ESC_ESC;
                break;
            default:
                mTxBuffer[offset++] = (byte)data;
                break;
            }
            if (offset >= BUFFER_SIZE) {
                mOutputStream.write(mTxBuffer, 0, offset);
                offset = 0;
            }
        }
        checksum = (-checksum) & 0xff;
        switch (checksum) {
        case FRAME:
            mTxBuffer[offset++] = ESC;
            mTxBuffer[offset++] = ESC_FRAME;
            break;
        case ESC:
            mTxBuffer[offset++] = ESC;
            mTxBuffer[offset++] = ESC_ESC;
            break;
        default:
            mTxBuffer[offset++] = (byte)checksum;
            break;
        }
        mTxBuffer[offset++] = FRAME;
        mOutputStream.write(mTxBuffer, 0, offset);
        mOutputStream.flush();
    }

    private class SendThread extends Thread {

        public void run() {
            try {
                byte control = 0;
                while (true) {
                    byte[] message = null;
                    switch (mState) {
                    case STATE_CLOSED:
                        return;
                    case STATE_OPEN_SENT:
                        sendMessage(EMPTY_MESSAGE, OPEN);
                        synchronized (mTxQueue) {
                            mTxQueue.wait();
                        }
                        break;
                    case STATE_OPEN_RECEIVED:
                        sendMessage(EMPTY_MESSAGE, OPEN_ACK);
                        synchronized (mTxQueue) {
                            mTxQueue.wait();
                        }
                        break;
                    case STATE_OPENED:
                        synchronized (mTxQueue) {
                            if ((mTxQueue[mNextTx] == null) || (mNextTx == mTxWindowClosedId))
                                mTxQueue.wait();
                            control = (byte)(ESTABLISHED_BIT | (mRxWindowClosedId << 4) | (mAckId << 2) | mNextTx);
                            // transmit message only if window is open
                            if (mNextTx != mTxWindowClosedId)
                                message = mTxQueue[mNextTx];
                            if (message != null) {
                                mNextTx++;
                                mNextTx &= 0x3;
                            }
                            else {
                                if (control == mLastTxControlValue)
                                    break;
                                // send ACK with no data
                                message = EMPTY_MESSAGE;
                            }
                        }
                        sendMessage(message, control);
                        break;
                    case STATE_CLOSING:
                        sendMessage(EMPTY_MESSAGE, CLOSED);
                        updateState(STATE_CLOSED);
                        return;
                    }
                }
            }
            catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }

    private class RetransmitThread extends Thread {
        public void run() {
            try {
                while (true) {
                    switch (mState) {
                    case STATE_CLOSED:
                        return;

                    case STATE_OPENED:
                        synchronized (mTxQueue) {
                            int index = (mLastTxAckd + 1) & 0x03;
                            if (mRetransmit[index]) {
                                mNextTx = index;
                                mTxQueue.notifyAll();
                                mRetransmitCount++;
                            }
                            else if (mTxQueue[index] != null)
                                mRetransmit[index] = true;
                        }
                        break;

                    default:
                        synchronized (mTxQueue) {
                            mTxQueue.notifyAll();
                        }
                        break;
                    }
                    try {
                        sleep(mRetransmitTime);
                    } catch (InterruptedException e) {}
                }
            }
            catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }

    private void readAvailable() throws IOException {
        mRxBufferIndex = 0;
        while (true) {
            mRxBufferCount = mInputStream.available();
            if (mRxBufferCount > 0) {
                if (mRxBufferCount > mRxBuffer.length)
                    mRxBufferCount = mRxBuffer.length;
                mInputStream.read(mRxBuffer, 0, mRxBufferCount);
                /*
                for (int i = 0; i < mRxBufferCount; i++) {
                    int data = mRxBuffer[i] & 0xff;
                    System.out.print(Integer.toHexString(data) + ' ');
                    if (data == FRAME)
                        System.out.println();
                }*/
                return;
            }
            try {
                Thread.sleep(mRxSleepTime);
                if (mState == STATE_CLOSED)
                    throw new MailboxStateException();
            }
            catch (InterruptedException e) {
            }
        }
    }

    private int receiveMessage(byte[] message) throws IOException {
        // read until a framing character is received
        do {
            if (mRxBufferIndex == mRxBufferCount)
                readAvailable();
        } while (mRxBuffer[mRxBufferIndex++] != FRAME);

        while (true) {
            int messageOffset = 0;
            // read until a non-framing character is received
            int control;
            do {
                if (mRxBufferIndex == mRxBufferCount)
                    readAvailable();
                control = mRxBuffer[mRxBufferIndex++] & 0xff;
            } while (control == FRAME);

            int checksum = control;

            // read until end of message (next framing character)
            boolean restart = false;
            do {
                if (mRxBufferIndex == mRxBufferCount)
                    readAvailable();
                int data = mRxBuffer[mRxBufferIndex++];

                switch (data) {
                case FRAME:
                    checksum &= 0xff;
                    if (checksum == 0)
                        return ((messageOffset - 1) << 8) | control;
                    else {
                        mReceiveErrors++;
                        restart = true;
                    }
                    break;
                case ESC:
                    if (mRxBufferIndex == mRxBufferCount)
                        readAvailable();
                    data = mRxBuffer[mRxBufferIndex++];

                    switch (data) {
                    case ESC_FRAME:
                        data = FRAME;
                        break;
                    case ESC_ESC:
                        data = ESC;
                        break;
                    default:
                        mReceiveErrors++;
                        restart = true;
                        break;
                    }
                    // fall through into default case
                default:
                    if (messageOffset < message.length)
                        message[messageOffset++] = (byte)data;
                    checksum += data;
                    break;
                }
            } while (!restart);
        }
    }

    /**
     * Gets the next message sent by the peer.  If no message is available, this
     * method will block until a message arrives.
     * @return
     *              a byte array containing the next message sent by the peer
     * @throws IOException
     */
    public byte[] getMessage() throws IOException {
        return getMessage(0);
    }

    /**
     * Gets the next message sent by the peer.  If no message is available, this
     * method will block until a message arrives or the maximum wait time has
     * been exceeded.
     *
     * @param   maximumWaitTime
     *              the maximum time in milliseconds to wait for a message
     *
     * @return
     *              a byte array containing the next message sent by the peer
     * @throws IOException
     */
    public byte[] getMessage(int maximumWaitTime) throws IOException {
        byte[] message;
        int newRxWindowClosedId;
        long waitUntil = 0;
        synchronized (mRxQueue) {
            while (mRxQueue[mNextGet] == null) {
                try {
                    if (maximumWaitTime == 0)
                        mRxQueue.wait();
                    else {
                        // the thread wakes up on any receive queue event, so timing
                        // is a bit more complicated
                        if (waitUntil == 0)
                            waitUntil = System.currentTimeMillis() + maximumWaitTime;

                        long thisWait = waitUntil - System.currentTimeMillis();
                        if (thisWait > 0)
                            mRxQueue.wait(thisWait);
                        else
                            throw new IOException("Timeout");
                    }
                    if (mState != STATE_OPENED)
                        throw new MailboxStateException();
                }
                catch (InterruptedException e) {}
            }

            message = mRxQueue[mNextGet];
            mRxQueue[mNextGet] = null;
            newRxWindowClosedId = mNextGet;
            mNextGet++;
            mNextGet &= 0x03;
            mRxQueue.notifyAll();
        }
        synchronized (mTxQueue) {
            mRxWindowClosedId = newRxWindowClosedId;
            mTxQueue.notifyAll();
        }
        return message;
    }

    private class ReceiveThread extends Thread {
        public void run() {
            try {
                byte[] message = null;
                while (true) {
                    if (message == null)
                        message = new byte[mMaxRxLength];
                    int lengthAndControl = receiveMessage(message);
                    int control = lengthAndControl & 0xff;
                    int length = lengthAndControl >> 8;
/*
                    mDebugStream.print("RX: " + Integer.toHexString(control));
                    if ((control & ESTABLISHED_BIT) != 0) {
                        System.out.print(
                            ", w: " + ((control >> 4) & 0x03) +
                            ", a: " + ((control >> 2) & 0x03) +
                            ", i: " + (control & 0x03));
                    }

                    if (length > 0) {
                        mDebugStream.print(", m: ");
                        for (int i = 0; i < length; ++i)
                            mDebugStream.print(Integer.toHexString(message[i] & 0xff) + ' ');
                    }
                    mDebugStream.println();
*/
                    switch (mState) {

                    case STATE_CLOSING:
                    case STATE_CLOSED:
                        return;

                    case STATE_OPEN_SENT:
                        if (control == OPEN_ACK)
                            updateState(STATE_OPENED);
                        else if (control == OPEN)
                            updateState(STATE_OPEN_RECEIVED);
                        break;

                    case STATE_OPEN_RECEIVED:
                        if ((control & (ESTABLISHED_BIT | 0x0c)) == ESTABLISHED_BIT) {
                            // ACK Id=0
                            updateState(STATE_OPENED);
                            // fall through to STATE_OPENED case
                        }
                        else if (control == OPEN_ACK) {
                            updateState(STATE_OPENED);
                            break;
                        }
                        else if (control == OPEN) {
                            // cause resend of OPEN_ACK
                            updateState(STATE_OPEN_RECEIVED);
                            break;
                        }
                        else {
                            close();
                            break;
                        }
                        // ACK Id=0 case falls through
                    case STATE_OPENED:
                        if ((control & ESTABLISHED_BIT) != 0) {
                            // ACK/DATA message
                            // process acknowledgements
                            synchronized (mTxQueue) {
                                mTxWindowClosedId = (lengthAndControl >> 4) & 0x03;
                                int ackId = (lengthAndControl >> 2) & 0x03;
                                while (mLastTxAckd != ackId) {
                                    mLastTxAckd++;
                                    mLastTxAckd &= 0x03;
                                    mTxQueue[mLastTxAckd] = null;
                                    mRetransmit[mLastTxAckd] = false;
                                }
                                mTxQueue.notifyAll();
                            }
                            synchronized (mRxQueue) {
                                // receive if id matches, length is non zero and rx queue isn't full
                                if (((control & 0x03) == mNextRx) && (length != 0)) {
                                    mAckId = mNextRx;
                                    while (mRxQueue[mNextRx] != null) {
                                        mRxQueue.wait();
                                        if (mState == STATE_CLOSED)
                                            return;
                                    }
                                    mRxQueue[mNextRx] = message;
                                    message = null;
                                    mNextRx++;
                                    mNextRx &= 0x03;
                                    mRxQueue.notifyAll();
                                }
                            }
                        }
                        else if (control == OPEN_ACK)
                            ack();
                        else
                            close();
                        break;
                    }
                }
            }
            catch (MailboxStateException e) {
                return;
            }
            catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }
}
